package com.song.shop.rocketmq;

import com.zeroturnaround.javarebel.con;
import org.apache.rocketmq.client.consumer.DefaultMQPullConsumer;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Song Yinghui
 * @date 2018/11/5
 * @Version 1.0
 */
@Component
public class RocketmqService {

    private Map<String, DefaultMQPushConsumer> mqCustomerMap = new HashMap<>();

    //@PostConstruct
    public void test() throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("DefaultCluster");
        DefaultMQPullConsumer pullConsumer = new DefaultMQPullConsumer();
        consumer.setNamesrvAddr("192.168.232.128:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
        //consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("songtopic", "*");
        consumer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
            list.forEach(item -> {
                System.out.println("--- 来着springboot自带的服务系统：" + new SimpleDateFormat("yyyyMMdd HH:mm:ss").format(new Date()) + "------");
                System.out.println(new String(item.getBody()));
            });
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        });
        /*consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) -> {
            consumeOrderlyContext.setAutoCommit(true);
            list.forEach( o -> {
                System.out.println("-----------");
                System.out.println(new String(o.getBody()));
            });
            return ConsumeOrderlyStatus.SUCCESS;
        });*/

        consumer.start();
    }


    //为每个Topic实现一个客户端
    public void subTopic(String topic, ICustomerCallBack callBack) throws MQClientException {
        DefaultMQPushConsumer customer;
        synchronized (mqCustomerMap) {
            customer = mqCustomerMap.get(topic);
            if (customer == null) {
                customer = new DefaultMQPushConsumer("DefaultCluster"+topic);
                customer.setNamesrvAddr("192.168.232.137:9876");
                customer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
                customer.setVipChannelEnabled(false);
                mqCustomerMap.put(topic , customer);
            }

            customer.subscribe(topic, "*");
            // 根据是否为严格顺序消费 还是集群消费 选择不同的listener
            customer.registerMessageListener((MessageListenerConcurrently) (list, context) -> {
                callBackHandle(list, callBack);
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            });
            customer.start();
        }
    }

    private void callBackHandle(List<MessageExt> list, ICustomerCallBack callBack) {
        for (MessageExt msg : list) {
            callBack.process(msg);
        }
    }
}
